package com.sise.netty.reactor模型.多Reactor模型;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.channels.spi.SelectorProvider;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author idea
 * @data 2019/4/6
 */
public class Processor {

    private ExecutorService service = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    private Selector selector;

    public Processor() throws IOException {
        this.selector = SelectorProvider.provider().openSelector();
        start();
    }

    public void addChannel(SocketChannel socketChannel) throws ClosedChannelException {
        socketChannel.register(this.selector, SelectionKey.OP_READ);
    }

    public void start() {
        service.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    System.out.println("----");
                    while (true) {
                        if (selector.selectNow() <= 0) {
                            continue;
                        } else {
                            Set<SelectionKey> set = selector.selectedKeys();
                            Iterator<SelectionKey> iterator = set.iterator();
                            while (iterator.hasNext()) {
                                SelectionKey sk = iterator.next();
                                if (sk.isReadable()) {
                                    ByteBuffer buffer = ByteBuffer.allocate(1024);
                                    SocketChannel socketChannel = (SocketChannel) sk.channel();
                                    int count = socketChannel.read(buffer);
                                    if (count < 0) {
                                        socketChannel.close();
                                        sk.cancel();
                                        System.out.println("读取结束");
                                        continue;
                                    } else {
                                        System.out.println("data读取" + new String(buffer.array()));
                                    }
                                }
                                iterator.remove();
                            }
                        }
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }

            }

        });
    }
}
